package com.lmy.rabbitmq.direct;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * direct路由模式 消费者接收消息
 * 已在RabbitMq的Web图形化界面中中将 direct_exchange和 queue1(routing key = “email”) queue3(routing key = "wechat") 两个队列绑定，这里只是使用channel接收指定队列消息
 */
public class Consumer {

    //这里启用多线程模拟多个消费者的场景
    private static Runnable runnable = new Runnable() {
        public void run() {
            //所有的中间件技术都是基于tcp/ip协议基础之上构建新型的协议规范，只不过rabbitmq遵循的是amqp
            //ip port
            //1.创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("169.254.27.100");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            //可以设置虚拟节点
            connectionFactory.setVirtualHost("/");
            Connection connection = null;
            Channel channel = null;
            //队列名称使用线程名称
            final String queueName = Thread.currentThread().getName();
            try {
                //2.创建连接Connection
                connection = connectionFactory.newConnection("生产者");
                //3.通过连接获取通道Channel
                channel = connection.createChannel();
                //4.接收消息，指定队列名进行消息接收
                channel.basicConsume(queueName, true, new DeliverCallback() {
                    @Override
                    public void handle(String s, Delivery delivery) throws IOException {
                        System.out.println(queueName + "接收到消息: " + new String(delivery.getBody(), "UTF-8"));
                    }
                }, new CancelCallback() {
                    @Override
                    public void handle(String s) throws IOException {
                        System.out.println(queueName + "消息接收失败.....");
                    }
                });
            }  catch (Exception e) {
                e.printStackTrace();
            } finally {
                //5.关闭通道
                if (channel != null && channel.isOpen()) {
                    try{
                        channel.close();
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
                //6.关闭连接
                if (connection != null && connection.isOpen()) {
                    try{
                        connection.close();
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            }
        }
    };

    public static void main(String[] args) {

        new Thread(runnable, "queue1").start();
        new Thread(runnable, "queue3").start();
    }
}
